package edu.vt.ece.rt.quorum.messsages.types;

import java.util.HashSet;
import java.util.List;
import java.util.PriorityQueue;
import java.util.Set;

import org.jgroups.Address;
import org.jgroups.Message;

import edu.vt.ece.rt.dstm.Context;
import edu.vt.ece.rt.dstm.ContextDelegator;
import edu.vt.ece.rt.dstm.QObject;
import edu.vt.ece.rt.dstm.TMProxy;
import edu.vt.ece.rt.quorum.ClustersManager;
import edu.vt.ece.rt.quorum.Resource;
import edu.vt.ece.rt.quorum.messsages.QMessage;
import edu.vt.ece.rt.quorum.messsages.QMessage.MessageType;
import edu.vt.ece.rt.quorum.util.Names;
import edu.vt.ece.rt.util.Logger.Logger;

/*
 * Commit Request Message to commit DSTM transaction
 * @author Sudhanshu
 */
public class CommitRequest extends QMessage implements Runnable {
	/*
	 * Comparing at local level can be incorrect as version update can be for
	 * this node leading to needless aborts
	 */
	Context cxt; // Context containing the Read-Write Set of objects
	int id; // Requester node of commit request
	Long parentId;

	public CommitRequest(Context c, int i, Long parent, String mId) {
		super(MessageType.CommitRequest);
		cxt = c;
		id = i;
		parentId = parent;
		messageId = mId;
	}
	

	public static boolean blast(Long txn, Context head) {
		Context hCxt = head;

		if (hCxt.getContextId().equals(txn)) {
			return true;
		} else {
			long t1 = txn / 10000000;
			long t2 = head.getContextId() / 10000000;
			
			if (t1 == t2) {
			
			Context child = ContextDelegator.getContext(txn);
			return Context.isParent(child, head);
			} else {
				return false;
			}
		}
	}
	
	@Override
	public void run() {
		synchronized (Resource.waitObj) {
			// Remove the Transaction from all PR and PW list

			cxt = ContextDelegator.getContext(cxt.getContextId());
			Logger.debug("Commit request from " + id + " for tnx "
					+ cxt.getContextId());
//			cxt.getCreateSet().addAll(cxt.getCreateSet());// error was here
			cxt.setParent(parentId); // registers the parent if not already done
//			Resource.displayObjects(cxt);


			Long txn = cxt.getContextId();
			Resource.rmFrLists(cxt);
			if (cxt.isAborted()) {
				try {
					Logger.debug("tnx " + txn
							+ " is already aborted, commit response abort");
					ClustersManager.getCluster(id + Names.writeQuorum).send(
							new Message(getSrcAddr(), new CommitResponse(null,
									txn, false)));
					Resource.incrementMessages(1);
					// send(new Message CommitResponse(src_addr, null, txn,
					// false));
				} catch (Exception e) {
					e.printStackTrace();
				}
				return;
			}

			// Create conflicting list
			Set<Context> conflictSet = new HashSet<Context>();

			// Add yourself in conflictSet
			conflictSet.add(cxt);
			HashSet<String> lockedObjects = new HashSet<String>();

			for (QObject wobj : cxt.getWriteSet()) {
				// Check local object status and version comparison
				if (Resource.getQObject(wobj.getId()).isProtected(txn)
						|| isStaleVersion(wobj)) {
					// True, means abort this transaction
					ContextDelegator.getContext(txn).setAborted();
					try {
						Logger.debug("Object " + wobj.getId()
								+ " is already locked by "
								+ Resource.getQObject(wobj.getId()).getOwner()
								+ " or stale, aborting tnx " + txn);
						ClustersManager.getCluster(id + Names.writeQuorum)
								.send(new Message(getSrcAddr(),
										new CommitResponse(null, txn, false)));
						Resource.incrementMessages(1);
						return;
					} catch (Exception e) {
						e.printStackTrace();
					} finally {
						// Unlock all previous taken objects
						unlockObjects(lockedObjects, txn);
					}
				}

				// Add PR and PW to conflicting list
				List<Context> l = Resource.getPRList(wobj.getId());
//				if (l != null) {
//					for (Context c : l) {
//						Logger.debug("PR list context is "+c.getContextId()+ " ");
//					}
//				}

				if (l != null)
					conflictSet.addAll(l);
				l = Resource.getPWList(wobj.getId());

//				if (l!=null) {
//					for (Context c : l) {
//						Logger.debug("PW list context is "+c.getContextId()+ " "+c.hashCode());
//					}
//				}
//				
				if (l != null)
					conflictSet.addAll(l);
				
				Logger.debug("size of Conflict set init is "+conflictSet.size());

				// Check if this transaction is highest priority transaction
				PriorityQueue<Context> prC = new PriorityQueue<Context>(
						conflictSet);
				Context hCxt = prC.peek();
				if (blast(cxt.getContextId(),hCxt)) { // Means still highest
														// priority

					if (cxt.isRoot()
							&& Resource.getQObject(wobj.getId())
									.lockObject(txn)) { // Now try lock
						lockedObjects.add(wobj.getId()); // If you get it add to
															// locked object
															// list
						Logger.debug("Locking for root transaction");
						Logger.debug(txn + " got lock on " + wobj.getId());
						continue;
					}

					if (cxt.isChild()) {
						Logger.debug("No locking for child transaction");
						continue;
					} else {
						Logger.debug("Object " + wobj.getId()
								+ " can not be locked, locked by "
								+ wobj.getOwner() + " aborting tnx " + txn);
					}
				}

				Logger.debug(txn + " is not highest Priority, highest is "
						+ hCxt.getContextId() + " getting aborted");
				// If not continued must be aborted
				ContextDelegator.getContext(txn).setAborted();
				try {
					ClustersManager.getCluster(id + Names.writeQuorum).send(
							new Message(getSrcAddr(), new CommitResponse(null,
									txn, false)));
					Resource.incrementMessages(1);
					return;
				} catch (Exception e) {
					e.printStackTrace();
				} finally {
					// Unlock all previous taken objects
					unlockObjects(lockedObjects, txn);
				}

			}
			Logger.debug("size of Conflict set after write set V is "+conflictSet.size());

			// Now check for the read Set
			for (QObject wobj : cxt.getReadSet()) {
				if (Resource.getQObject(wobj.getId()).isProtected(txn)
						|| isStaleVersion(wobj)) {
					// True, means abort this transaction
					ContextDelegator.getContext(txn).setAborted();
					try {
						ClustersManager.getCluster(id + Names.writeQuorum)
								.send(new Message(getSrcAddr(),
										new CommitResponse(null, txn, false)));
						Resource.incrementMessages(1);
						return;
					} catch (Exception e) {
						e.printStackTrace();
					} finally {
						// Unlock all previous taken objects
						unlockObjects(lockedObjects, txn);
					}
				}

				// Add PW to conflicting list
				List<Context> l = Resource.getPWList(wobj.getId());
				if (l != null)
					conflictSet.addAll(l);

				// Check if this transaction is highest priority transaction
				PriorityQueue<Context> prC = new PriorityQueue<Context>(
						conflictSet);
				if (blast(cxt.getContextId(), prC.peek())) // Means still
															// highest priority
					continue;

				// If not continued must be aborted
				ContextDelegator.getContext(txn).setAborted();
				try {
					Logger.debug("Read Validation Failed for" + wobj
							+ " aborting tnx " + txn);
					ClustersManager.getCluster(id + Names.writeQuorum).send(
							new Message(getSrcAddr(), new CommitResponse(null,
									txn, false)));
					Resource.incrementMessages(1);
					return;
				} catch (Exception e) {
					e.printStackTrace();
				} finally {
					// Unlock all previous taken objects
					unlockObjects(lockedObjects, txn);
				}
			}
			Logger.debug("size of Conflict set after read set V is "+conflictSet.size());

			// Check if Context got aborted
			if (ContextDelegator.getContext(txn).isAborted()) {
				try {
					ClustersManager.getCluster(id + Names.writeQuorum).send(
							new Message(getSrcAddr(), new CommitResponse(null,
									txn, false)));
					Resource.incrementMessages(1);
					return;
				} catch (Exception e) {
					e.printStackTrace();
				} finally {
					// Unlock all previous taken objects
					unlockObjects(lockedObjects, txn);
				}
			}

			// If passed the read and write both validation, it should be
			// committed
			// remove yourself from conflicting list
			conflictSet.remove(cxt);
			Logger.debug("size of Conflict set before is "+conflictSet.size());
			for (Context c : conflictSet) {
				Logger.debug("context is "+c.getContextId());
			}
			while (conflictSet.contains(cxt))
					conflictSet.remove(cxt);
			Logger.debug("size of Conflict set is "+conflictSet.size());
			// Abort all the other transaction in conflict list
			if (ContextDelegator.getContext(txn).isRoot()) {
				for (Context c : conflictSet) {
					if (!blast(cxt.getContextId(), c))
						c.setAborted();
				}
			}
			
			CommitResponse ca = new CommitResponse(conflictSet, txn, true);
			try {
				Logger.debug("Successfully verified tnx " + txn);
				ClustersManager.getCluster(id + Names.writeQuorum).send(
						new Message(getSrcAddr(), ca));
				Resource.incrementMessages(ClustersManager.getCluster(id + Names.writeQuorum).currentView.size());
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * @param lockedObjects
	 */
	private void unlockObjects(HashSet<String> lockedObjects, Long txn) {

		for (String id : lockedObjects) {
			Resource.getQObject(id).unlockObject(txn);
			Logger.debug(txn + " left the lock on " + id);
		}
	}

	/**
	 * Stale Version
	 * 
	 * @param wobj
	 * @param cxt
	 * @return
	 */
	private boolean isStaleVersion(QObject wobj) {
		boolean b = wobj.getVersion() < Resource.getQObject(wobj.getId())
				.getVersion();
		Logger.debug("Object is stale=" + b);
		return b;
	}
}
